package org.niit.service

import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
import org.apache.spark.sql.functions._
import org.apache.spark.streaming.dstream.DStream
import org.niit.bean.Order
import org.niit.dao.RealtimeAnalysisDao
import org.niit.util.ConfigUtil

/**
 * 实时分析服务，使用Spark Streaming进行实时数据处理
 */
object RealtimeAnalysisService {

  /**
   * 处理实时订单数据
   *
   * @param spark SparkSession
   * @param orderStream 订单数据流
   */
  def processRealtimeOrders(spark: SparkSession, orderStream: DStream[String]): Unit = {
    // 转换RDD为DataFrame进行分析
    orderStream.foreachRDD { rdd =>
      if (!rdd.isEmpty()) {
        try {
          import spark.implicits._

          // 解析订单数据
          val orderDF = rdd.map { line =>
            val fields = line.split(",", -1)
            (
              fields(0),  // 订单ID
              fields(1),  // 用户ID
              fields(2).toInt,  // 年龄
              fields(3),  // 性别
              fields(4),  // 商铺名称
              fields(5),  // 菜品名称
              fields(6),  // 菜品类别
              fields(7).toDouble,  // 价格
              fields(8),  // 城市
              fields(9),  // 时间段
              fields(10),  // 平台
              fields(11).toDouble,  // 评分
              fields(12)  // 下单时间
            )
          }.toDF("订单ID", "用户ID", "年龄", "性别", "商铺名称", "菜品名称", "菜品类别", "价格", "城市", "时间段", "平台", "评分", "下单时间")
          
          // 针对此批次进行各维度分析
          analyzeRealtimeOrderStats(orderDF)
          analyzeRealtimeRegion(orderDF)
          analyzeRealtimeCategory(orderDF)
          analyzeRealtimeTimePeriod(orderDF)
          analyzeRealtimeMerchant(orderDF)
          analyzeRealtimePlatform(orderDF)
          
          println(s"成功处理 ${orderDF.count()} 条订单数据")
        } catch {
          case e: Exception => 
            println("处理实时订单数据时发生异常")
            e.printStackTrace()
        }
      }
    }
  }

  /**
   * 分析实时订单统计数据
   *
   * @param orderDF 订单DataFrame
   */
  private def analyzeRealtimeOrderStats(orderDF: DataFrame): Unit = {
    // 统计订单总量
    val totalOrders = orderDF.count()
    
    // 计算每分钟订单数（假设数据是最近一分钟的）
    val ordersPerMinute = totalOrders.toDouble
    
    // 计算平均订单金额
    val avgAmount = orderDF.agg(avg("价格").alias("avg_price")).first().getDouble(0)
    
    // 计算活跃城市数量
    val activeCities = orderDF.select("城市").distinct().count().toInt
    
    // 保存到数据库
    RealtimeAnalysisDao.saveOrderStats(totalOrders.toInt, ordersPerMinute, avgAmount, activeCities)
    
    println(s"实时订单统计: 总量=$totalOrders, 每15秒=$ordersPerMinute, 平均金额=$avgAmount, 活跃城市=$activeCities")
  }

  /**
   * 分析实时地区分布
   *
   * @param orderDF 订单DataFrame
   */
  private def analyzeRealtimeRegion(orderDF: DataFrame): Unit = {
    // 按城市分组统计订单数量
    val regionDF = orderDF.groupBy("城市")
      .agg(count("*").alias("order_count"))
      .orderBy(desc("order_count"))
    
    // 计算总订单数和百分比
    val totalOrders = orderDF.count()
    val resultDF = regionDF.withColumn("percentage", (col("order_count") / totalOrders) * 100)
    
    // 清空旧数据
    //RealtimeAnalysisDao.clearRealtimeTable("realtime_region")
    
    // 保存到数据库
    resultDF.collect().foreach { row =>
      val city = row.getString(0)
      val orderCount = row.getLong(1).toInt
      val percentage = row.getDouble(2)
      RealtimeAnalysisDao.saveRealtimeRegion(city, orderCount, percentage)
    }
  }

  /**
   * 分析实时菜品类别分布
   *
   * @param orderDF 订单DataFrame
   */
  private def analyzeRealtimeCategory(orderDF: DataFrame): Unit = {
    // 按菜品类别分组统计订单数量
    val categoryDF = orderDF.groupBy("菜品类别")
      .agg(count("*").alias("order_count"))
      .orderBy(desc("order_count"))
    
    // 计算总订单数和百分比
    val totalOrders = orderDF.count()
    val resultDF = categoryDF.withColumn("percentage", (col("order_count") / totalOrders) * 100)
    
    // 清空旧数据
    //RealtimeAnalysisDao.clearRealtimeTable("realtime_category")
    
    // 保存到数据库
    resultDF.collect().foreach { row =>
      val category = row.getString(0)
      val orderCount = row.getLong(1).toInt
      val percentage = row.getDouble(2)
      RealtimeAnalysisDao.saveRealtimeCategory(category, orderCount, percentage)
    }
  }

  /**
   * 分析实时活跃时段
   *
   * @param orderDF 订单DataFrame
   */
  private def analyzeRealtimeTimePeriod(orderDF: DataFrame): Unit = {
    // 按时间段分组统计订单数量
    val timePeriodDF = orderDF.groupBy("时间段")
      .agg(count("*").alias("order_count"))
    
    // 计算总订单数和百分比
    val totalOrders = orderDF.count()
    val resultDF = timePeriodDF.withColumn("percentage", (col("order_count") / totalOrders) * 100)
    
    // 清空旧数据
    //RealtimeAnalysisDao.clearRealtimeTable("realtime_time_period")
    
    // 保存到数据库
    resultDF.collect().foreach { row =>
      val timePeriod = row.getString(0)
      val orderCount = row.getLong(1).toInt
      val percentage = row.getDouble(2)
      RealtimeAnalysisDao.saveRealtimeTimePeriod(timePeriod, orderCount, percentage)
    }
  }

  /**
   * 分析实时商铺订单
   *
   * @param orderDF 订单DataFrame
   */
  private def analyzeRealtimeMerchant(orderDF: DataFrame): Unit = {
    // 按商铺名称分组统计订单数量
    val merchantDF = orderDF.groupBy("商铺名称")
      .agg(count("*").alias("order_count"))
      .orderBy(desc("order_count"))
      .limit(10)  // 只取前10名
    
    // 计算总订单数和百分比
    val totalOrders = orderDF.count()
    val resultDF = merchantDF.withColumn("percentage", (col("order_count") / totalOrders) * 100)
    
    // 清空旧数据
    //RealtimeAnalysisDao.clearRealtimeTable("realtime_merchant")
    
    // 保存到数据库
    resultDF.collect().foreach { row =>
      val merchantName = row.getString(0)
      val orderCount = row.getLong(1).toInt
      val percentage = row.getDouble(2)
      RealtimeAnalysisDao.saveRealtimeMerchant(merchantName, orderCount, percentage)
    }
  }

  /**
   * 分析实时平台分布
   *
   * @param orderDF 订单DataFrame
   */
  private def analyzeRealtimePlatform(orderDF: DataFrame): Unit = {
    // 按平台分组统计订单数量
    val platformDF = orderDF.groupBy("平台")
      .agg(count("*").alias("order_count"))
      .orderBy(desc("order_count"))
    
    // 计算总订单数和百分比
    val totalOrders = orderDF.count()
    val resultDF = platformDF.withColumn("percentage", (col("order_count") / totalOrders) * 100)
    
    // 清空旧数据
    //RealtimeAnalysisDao.clearRealtimeTable("realtime_platform")
    
    // 保存到数据库
    resultDF.collect().foreach { row =>
      val platform = row.getString(0)
      val orderCount = row.getLong(1).toInt
      val percentage = row.getDouble(2)
      RealtimeAnalysisDao.saveRealtimePlatform(platform, orderCount, percentage)
    }
  }
} 